//package com.open.source.kafka;
//
//import lombok.extern.slf4j.Slf4j;
//import org.apache.kafka.clients.consumer.ConsumerConfig;
//import org.springframework.beans.BeansException;
//import org.springframework.boot.ApplicationArguments;
//import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
//import org.springframework.context.ApplicationContext;
//import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
//import org.springframework.kafka.listener.ContainerProperties;
//import org.springframework.kafka.listener.ErrorHandler;
//import org.springframework.kafka.listener.GenericMessageListenerContainer;
//import org.springframework.kafka.listener.MessageListener;
//
///**
// * 手动提交
// *
// * @author ZonLen since on 2022/7/29 下午12:22
// */
//
//@Slf4j
//public abstract class AbstractKafkaDltConsumerListener implements KafkaConsumerListener,
//    MessageListener<String, String>, ErrorHandler {
//
////  private static final int MAX_RETRY = 3;
////
////  private static final BiFunction<ConsumerRecord<?, ?>, Exception, TopicPartition> FUNCTION = (record, exp) -> {
////    final Headers headers = record.headers();
////    final Header header = headers.lastHeader(MqConstant.X_RETRIES_HEADER);
////    final int retryTime = Integer
////        .parseInt(new String(header.value()));
////    if (retryTime >= MAX_RETRY) {
////      return new TopicPartition(record.topic() + ".ERROR", record.partition());
////    }
////    return new TopicPartition(record.topic() + ".DLT", record.partition());
////  };
//
//  protected ApplicationContext applicationContext;
//
//  private KafkaProperties kafkaProperties;
//
////  @Override
////  public ConsumerFactory<Object, Object> consumerFactory() {
////    final Map<String, Object> consumerProperties = ApplicationContextHolder
////        .getBean(KafkaProperties.class).buildConsumerProperties();
////    consumerProperties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId());
////    consumerProperties
////        .put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
////    consumerProperties
////        .put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
////    consumerProperties.put("value.deserializer.encoding", deserializerEncoding());
//////    consumerProperties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
////    consumerProperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
////    return new DefaultKafkaConsumerFactory<>(consumerProperties);
////  }
//
//
//  /**
//   * 设置默认消费者组
//   */
//  public String groupId() {
//    return kafkaProperties.buildConsumerProperties()
//        .get(ConsumerConfig.GROUP_ID_CONFIG).toString();
//  }
//
//  protected GenericMessageListenerContainer<String, String> genericMessageListenerContainer;
//
//
//  @Override
//  public void destroy() {
//    if (kafkaProperties == null) {
//      return;
//    }
//    genericMessageListenerContainer.destroy();
//    log.info("kafka consumer topic <{}> manual commit listener stop", String.join(",", kafkaTopic()));
//  }
//
//  @Override
//  public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
//    this.applicationContext = applicationContext;
//  }
//
//  @Override
//  public void run(ApplicationArguments args) {
//    try {
//      kafkaProperties = applicationContext.getBean(KafkaProperties.class);
//    } catch (BeansException e) {
//      log.info("Kafka topic<{}> custom listener<{}> Is not enabled", String.join(",", kafkaTopic()),
//          getClass().getName());
//      return;
//    }
//    String topics = kafkaTopic();
//    if (null == topics || topics.length == 0) {
//      throw new RuntimeException("Non topics listener, please check");
//    }
//    ContainerProperties containerProperties = new ContainerProperties(topics);
//    containerProperties.setMessageListener(this);
////    containerProperties.setAckMode(AckMode.MANUAL_IMMEDIATE);
//    ConcurrentMessageListenerContainer<String, String> concurrentMessageListenerContainer = new ConcurrentMessageListenerContainer<>(
//        consumerFactory(), containerProperties);
//
//    concurrentMessageListenerContainer.start();
//    this.genericMessageListenerContainer = concurrentMessageListenerContainer;
//  }
//
//}
